#include "cgzCTcpAsioS.h"
#include "cgzCThreadPool.h"

cgzSessionS::cgzSessionS(
    int fd,
    int ufd,
    tcp::socket socket,
    cgzCTcpAsioS *that)
    : cgzCTcpFd(fd, ufd),
      m_socket(std::move(socket)),
      m_that(that)
{
    m_ip = m_socket.remote_endpoint().address().to_string();
    m_port = m_socket.remote_endpoint().port();
}

void cgzSessionS::do_read()
{
    m_socket.async_read_some(buffer(m_recv.head(), m_recv.dire()),
                             [this](boost::system::error_code ec, size_t tlen)
                             {
                                 auto that = this->m_that;
                                 if (!ec)
                                 {
                                     std::unique_lock lock(that->m_rwmutex);
                                     m_recv.push(tlen);
                                     while (that->sticking(static_cast<cgzCTcpFd *>(this)))
                                         ;
                                     do_read();
                                 }
                                 else
                                 {
                                     cgz_that_error("err:{}", ec.message().c_str());
                                 }
                             });
}

void cgzSessionS::do_send(cgzCSendS *send)
{
    auto leftlen = send->m_len - send->m_sendlen;
    auto leftptr = send->m_data + send->m_sendlen;
    async_write(m_socket, buffer(leftptr, leftlen),
                [this, send](boost::system::error_code ec, std::size_t)
                {
                    if (ec)
                    {
                        auto that = this->m_that;
                        that->disstcp(this);
                        cgz_that_error("err:{}", ec.message().c_str());
                    }
                    send->~cgzCSendS();
                    free(send);
                });
}

cgzCTcpAsioS::cgzCTcpAsioS(cgzCThreadPool *pool,
                           const char *name,
                           const std::string ip,
                           const int port,
                           const int cid)
    : m_iocontext(nullptr),
      m_acceptor(nullptr),
      cgzCTcpS(pool, name, ip, port, cid)

{
    m_tenum = cgzETask::CAsioTcpS;
}

cgzCTcpAsioS::~cgzCTcpAsioS()
{
    m_iocontext->stop();
    m_iocontext = nullptr;

    m_acceptor->close();
    m_acceptor = nullptr;

    m_thread->join();
    m_thread = nullptr;
}

bool cgzCTcpAsioS::tcpbind()
{
    m_iocontext = new io_context();
    m_acceptor = new ip::tcp::acceptor(*m_iocontext, tcp::endpoint(tcp::v4(), m_port));
    // 设置socket选项
    m_acceptor->set_option(tcp::no_delay(true));
    m_acceptor->set_option(socket_base::reuse_address(true));
    // 获取socket句柄
    m_sfd = m_acceptor->native_handle();
    // 加入线程池处理
    m_pool->addTask(m_name.c_str(), this);
    // 创建工作线程
    m_thread = new std::thread(&cgzCTcpAsioS::do_tgoing, this);
    m_pool->push_thread(m_thread);
    return true;
}

void cgzCTcpAsioS::do_tgoing()
{
    do_accept();
    while (m_bRunning)
    {
        m_iocontext->run();
    }
}

bool cgzCTcpAsioS::tcpread()
{
    assert(false);
    return false;
}

bool cgzCTcpAsioS::tcpsend()
{
    while (!m_sendqueue.empty())
    {
        auto csend = m_sendqueue.front();
        m_sendqueue.pop();
        auto it = m_ufdmap.find(csend->m_ufd);
        if (it == m_ufdmap.end())
        {
            // 已经失效的客户端-删除发送请求
            csend->~cgzCSendS();
            free(csend);
            continue;
        }

        auto fdinfo = static_cast<cgzSessionS *>(it->second);
        fdinfo->do_send(csend);
    }
    return true;
}

bool cgzCTcpAsioS::taccpet()
{
    assert(false);
    return false;
}

void cgzCTcpAsioS::do_accept()
{
    m_acceptor->async_accept(
        [this](boost::system::error_code ec, tcp::socket socket)
        {
            if (!ec)
            {
                // 创建新的连接
                int fd = socket.native_handle();
                int ufd = ++m_ufd;
                auto node = malloc(sizeof(cgzSessionS));
                auto finf = new (node) cgzSessionS(fd, ufd, std::move(socket), this);
                finf->m_hand = true;

                // 保存连接信息
                m_ufdmap.insert(std::make_pair(ufd, finf));
                m_fdmap.insert(std::make_pair(fd, finf));
                finf->do_read();
                // 发送连接消息
                sendaccept(finf);
                cgz_this_error("accect:{} ip:{} port:{}", m_name.c_str(), finf->m_ip.c_str(), finf->m_port);
            }
            else
            {
                cgz_this_error("name:{} err:{}", m_name.c_str(), ec.message().c_str());
            }
            do_accept(); // 继续接受新连接
        });
}